package com.bank.backend.boot.server.kafka;

import com.alibaba.fastjson2.JSON;
import com.bank.backend.boot.base.enums.TradeEnum;
import com.bank.backend.boot.base.model.dto.ModelDTO;
import com.bank.backend.boot.base.model.dto.ModelResultDTO;
import com.bank.backend.boot.base.model.dto.ModelTradeDTO;
import com.bank.backend.boot.base.model.entity.Trade;
import com.bank.backend.boot.base.model.vo.ModelVO;
import com.bank.backend.boot.server.feign.ModelFeign;
import com.bank.backend.boot.server.mapper.TradeMapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author jiaowei
 * @since 2024/11/18 13:15
 */
@Component
@Slf4j
public class ConsumerKafka {

    @Resource
    private ModelFeign modelFeign;

    @Resource
    private TradeMapper tradeMapper;

    @KafkaListener(topics = "bank-backend-trade", groupId = "kafka")
    public void consumerTrade(ConsumerRecord<Integer, String> record) {
        log.info("consumer message: {}", record.value());
    }

    // TODO 识别欺诈算法
    @KafkaListener(topics = "bank-backend-identify", groupId = "kafka")
    public void consumerIdentify(ConsumerRecord<Integer, String> record) {
        log.info("consumer message: {}", record.value());
        ModelTradeDTO modelTradeDTO = JSON.parseObject(record.value(), ModelTradeDTO.class);

        Trade trade = modelTradeDTO.getTrade();
        ModelDTO modelDTO = modelTradeDTO.getModelDTO();

        ModelVO modelVO = getModelVO(trade, modelDTO);
        ModelResultDTO modelResultDTO = modelFeign.detectFraud(modelVO);
        log.warn("modelResultDTO:{}", modelResultDTO);
        updateTrade(trade, modelResultDTO, modelDTO);
    }

    private void updateTrade(Trade trade, ModelResultDTO modelResultDTO,  ModelDTO modelDTO) {
        UpdateWrapper<Trade> updateWrapper = new UpdateWrapper<>();
        updateWrapper.eq("id", trade.getId());
        if (modelResultDTO.getPredictedClass() == 1) {
            updateWrapper.set("risk", TradeEnum.RISK_3.getCode())
                        .set("finish_time", new Date())
                        .set("finish", TradeEnum.Finish_2.getCode())
                        .set("probability", 0.9)
                        .set("description", modelDTO.getTransactionDescription());

        } else {
            double probability = modelResultDTO.getProbability();
            updateWrapper.set("description", modelDTO.getTransactionDescription());
            if (probability < 0.7) {
                updateWrapper.set("risk", TradeEnum.RISK_3.getCode())
                        .set("probability", probability + 0.2);
            } else if (probability < 0.8) {
                updateWrapper.set("risk", TradeEnum.RISK_2.getCode())
                        .set("probability", probability);
            } else if (probability < 0.9) {
                updateWrapper.set("risk", TradeEnum.RISK_1.getCode())
                        .set("probability", probability);
            } else {
                updateWrapper.set("finish", TradeEnum.Finish_1.getCode())
                        .set("finish_time", new Date())
                        .set("probability", probability);
            }
        }
        int i = tradeMapper.update(updateWrapper);
    }

    private static ModelVO getModelVO(Trade trade, ModelDTO modelDTO) {
        ModelVO modelVO = new ModelVO();
        modelVO.setCustomerId(String.valueOf(trade.getId()));
        modelVO.setGender(modelDTO.getGender());
        modelVO.setAge(modelDTO.getAge());
        modelVO.setState(modelDTO.getState());
        modelVO.setCity(modelDTO.getCity());
        modelVO.setBankBranch(modelDTO.getBankBranch());
        modelVO.setAccountType(modelDTO.getAccountType());
        Date date = trade.getTradeTime();
        SimpleDateFormat day = new SimpleDateFormat("dd-MM-yyyy");
        String nowDay = day.format(new Date());
        modelVO.setTransactionDate(nowDay);
        SimpleDateFormat time = new SimpleDateFormat("HH:mm:ss");
        String nowTime = time.format(new Date());
        modelVO.setTransactionTime(nowTime);
        modelVO.setTransactionAmount(modelDTO.getTransactionAmount());
        modelVO.setMerchantId("*************");
        modelVO.setTransactionType(modelDTO.getTransactionType());
        modelVO.setMerchantCategory(modelDTO.getMerchantCategory());
        modelVO.setAccountBalance(modelDTO.getAccountBalance());
        modelVO.setTransactionDevice(modelDTO.getTransactionDevice());
        modelVO.setTransactionLocation(modelDTO.getTransactionLocation());
        modelVO.setDeviceType(modelDTO.getDeviceType());
        modelVO.setTransactionDescription(modelDTO.getTransactionDescription());
        log.warn("modelVO:{}", modelVO);
        return modelVO;
    }
}
